package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;
import hit.edu.zjc.parsers.ByteTupleParserFactory;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;

public class SourceHashSetIntegrationOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private String[] FilePaths;//含有HashSet的二进制的文件路径
    // ArrayList<HashSet<Integer>> SetList=new ArrayList<HashSet<Integer>>();
    public SourceHashSetIntegrationOperatorDescriptor(IOperatorDescriptorRegistry spec,String[] files) {
        super(spec, 0, 1);  
        recordDescriptors[0]  = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        FilePaths=files;
    }

    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
    	 final ITupleParser parser =new ByteTupleParserFactory().createTupleParser(ctx);
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {
        	
        	   /*
        	    *  将第二个SetList，合并到第一SetList上。
        	    * */
        	  public void MergeTwoSetList(ArrayList<HashSet<Integer>> SetList,ArrayList<HashSet<Integer>> otherSetList)
        	  {
        		  for(int i=0;i<otherSetList.size();i++)
        		  {
        			  HashSet<Integer> otherset=otherSetList.get(i);
        			  HashSet<Integer> set=null;
        			  ArrayList<Integer> CoverIndex=new ArrayList<Integer>();
        			  int j=0;
        			  for(j=0;j<SetList.size();j++)
        			  {
        				  set=SetList.get(j);
        				  boolean iscover=isCover(otherset,set);
        				  if(iscover)
        				  {
        					  CoverIndex.add(0,j);//构成降序列
        				  }
        			  }
        			  if(CoverIndex.size()==0)//不相交
        			  {
        				  SetList.add(otherset);
        			  }
        			  else
        			  {
        				  for(int m=0;m<CoverIndex.size();m++)
        				  {
        					  int index=CoverIndex.get(m);
        					  otherset.addAll(SetList.get(index));
        					  SetList.remove(index);
        				  }
        				  SetList.add(otherset);
        			  }
        		  }
        	  }
        	  //判断两个集合是否相交
              public boolean isCover(HashSet<Integer>set1,HashSet<Integer>set2)
                 {
              	   Iterator<Integer>iterator=set1.iterator();
            	   while(iterator.hasNext())
            	   		{
            		   Integer data=iterator.next();
            		   boolean res=set2.contains(data);
            		   if(res)
            			   return true;
            	   		}
            	   return false;
                 }

        	
            @Override
            public void initialize() throws HyracksDataException {

            	final int filelength=FilePaths.length;
            	ArrayList<HashSet<Integer>> SetList=(ArrayList<HashSet<Integer>>)RWObject.ReadObject(FilePaths[0]);
            	//把其他HashSet合并到第一个上。
            	for(int i=1;i<filelength;i++)
            	{
            		ArrayList<HashSet<Integer>> OtherSetList=(ArrayList<HashSet<Integer>>)RWObject.ReadObject(FilePaths[i]);
            		MergeTwoSetList(SetList,OtherSetList);
            	}
            	//合并结束。
      	
             writer.open();
             try
             	{
	     	    	byte[] SetListBytes =ObjectConvert.ObjectToByteArray(SetList);  		   
	     		   ByteArrayInputStream bais = new ByteArrayInputStream(SetListBytes);	
	     	    	parser.parse(bais, writer);
	     	      bais.close();
	            
             	}
             catch(Exception e)
             	{
            	 	writer.fail();
            	 	e.printStackTrace();
             	}
             finally
             	{
            	 	writer.close();
             	}
            }
        };
    }
}